package com.fanshuai.spring.boot.component;

import com.fanshuai.spring.boot.component.handler.MessageHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.MessageListenerContainer;

import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

@Slf4j
public class KafkaConsumerEngine implements Closeable {
    private List<MessageListenerContainer> listenerContainers = new ArrayList<>();

    public KafkaConsumerEngine(ConsumerFactory<String, String> consumerFactory, List<MessageHandler> messageHandlerList, GenericKafkaListener kafkaListener) {
        if (CollectionUtils.isNotEmpty(messageHandlerList)) {
            for (MessageHandler handler: messageHandlerList) {
                ContainerProperties properties = new ContainerProperties(handler.topics().toArray(new String[0]));
                properties.setMessageListener(kafkaListener);

                MessageListenerContainer container = new KafkaMessageListenerContainer<>(consumerFactory, properties);
                listenerContainers.add(container);
            }
        }

        buildMessageContainers();
    }

    private void buildMessageContainers() {
        if (CollectionUtils.isNotEmpty(listenerContainers)) {
            for (MessageListenerContainer container : listenerContainers) {
                container.start();
                log.info("container start, id={}", container.getListenerId());
            }
        }
    }

    @Override
    public void close() throws IOException {
        if (CollectionUtils.isNotEmpty(listenerContainers)) {
            for (MessageListenerContainer container : listenerContainers) {
                container.stop();
                log.info("container stop, id={}", container.getListenerId());
            }
        }
    }
}
